package com.lambert.ginkgo.mock.controller;


import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.UUID;

@RestController
public class StreamController {

    /**
     * 流式接口：实时推送数据（每1秒推送一次）
     * 客户端通过 SSE 接收，Content-Type 为 text/event-stream
     */
    @GetMapping(value = "/stream/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamData(@PathVariable String userId) {
        // 模拟每秒生成一条数据（实际场景可替换为数据库查询、消息队列消费等）
        return Flux.interval(Duration.ofSeconds(1))
                .map(sequence -> {
                    // 构建 SSE 事件（包含数据、事件ID、事件类型等）
                    return ServerSentEvent.<String>builder()
                            .id(UUID.randomUUID().toString()) // 事件唯一ID（用于重连时定位断点）
                            .event("data-update") // 事件类型（客户端可根据类型处理）
                            .data("用户 " + userId + " 的实时数据：" + sequence) // 实际推送的数据
                            .build();
                })
                .take(10); // 示例：仅推送10条数据后结束（实际场景可移除，实现无限流）
    }

    /**
     * 流式接口：推送数据库实时变更（模拟）
     */
    @GetMapping(value = "/db/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamDbChanges() {
        // 实际场景中，可替换为监听数据库 binlog 或消息队列（如 Kafka）的数据流
        return Flux.generate(sink -> {
            try {
                // 模拟数据库变更（实际应是异步接收变更事件）
                String change = "数据库新增记录：" + System.currentTimeMillis();
                sink.next(ServerSentEvent.<String>builder()
                        .data(change)
                        .build());
                Thread.sleep(2000); // 每2秒推送一次
            } catch (InterruptedException e) {
                sink.error(e);
            }
        });
    }
}
